package com.jorado.pulsar.client;

import com.jorado.pulsar.client.metrics.PulsarMetricsRecorder;
import com.jorado.pulsar.client.trace.TracingConsumerInterceptor;
import com.jorado.pulsar.client.util.GetAppNameUtil;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.ConsumerBuilderImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.shade.org.apache.commons.lang3.StringUtils;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

public class ConsumerCreator<T> extends ConsumerBuilderImpl<T> {

    protected boolean enableMetricsCollect = true;
    protected boolean enableTracing = false;
    protected String projectName = "default";
    protected TracingConsumerInterceptor<T> traceInterceptor;

    public ConsumerCreator(PulsarClientImpl client, Schema schema) {
        super(client, schema);
    }

    @Override
    public ConsumerCreator<T> clone() {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<Consumer<T>> subscribeAsync() {
        CompletableFuture<Consumer<T>> future = super.subscribeAsync();
        future.whenComplete((consumer, throwable) -> {
            if (enableMetricsCollect && throwable == null) {
                PulsarMetricsRecorder.getInstance().registerConsumer(consumer);
            }

        });
        return future;
    }

    @Override
    public ConsumerCreator<T> topic(String... topicNames) {
        super.topic(topicNames);
        return this;
    }

    @Override
    public ConsumerCreator<T> topics(List topicNames) {
        super.topics(topicNames);
        return this;
    }

    @Override
    public ConsumerCreator<T> topicsPattern(Pattern topicsPattern) {
        super.topicsPattern(topicsPattern);
        return this;
    }

    @Override
    public ConsumerCreator<T> topicsPattern(String topicsPattern) {
        super.topicsPattern(topicsPattern);
        return this;
    }

    @Override
    public ConsumerCreator<T> subscriptionName(String subscriptionName) {
        super.subscriptionName(subscriptionName);
        return this;
    }

    @Override
    public ConsumerCreator<T> ackTimeout(long ackTimeout, TimeUnit timeUnit) {
        super.ackTimeout(ackTimeout, timeUnit);
        return this;
    }

    @Override
    public ConsumerCreator<T> subscriptionType(SubscriptionType subscriptionType) {
        super.subscriptionType(subscriptionType);
        return this;
    }

    @Override
    public ConsumerCreator<T> messageListener(MessageListener messageListener) {
        super.messageListener(messageListener);
        return this;
    }


    @Override
    public ConsumerCreator<T> receiverQueueSize(int receiverQueueSize) {
        super.receiverQueueSize(receiverQueueSize);
        return this;
    }

    @Override
    public ConsumerCreator<T> acknowledgmentGroupTime(long delay, TimeUnit unit) {
        super.acknowledgmentGroupTime(delay, unit);
        return this;
    }

    @Override
    public ConsumerCreator<T> consumerName(String consumerName) {
        super.consumerName(consumerName);
        return this;
    }


    @Override
    public ConsumerCreator<T> maxTotalReceiverQueueSizeAcrossPartitions(int maxTotalReceiverQueueSizeAcrossPartitions) {
        super.maxTotalReceiverQueueSizeAcrossPartitions(maxTotalReceiverQueueSizeAcrossPartitions);
        return this;
    }

    @Override
    public ConsumerCreator<T> readCompacted(boolean readCompacted) {
        super.readCompacted(readCompacted);
        return this;
    }

    @Override
    public ConsumerCreator<T> patternAutoDiscoveryPeriod(int periodInMinutes) {
        super.patternAutoDiscoveryPeriod(periodInMinutes);
        return this;
    }

    @Override
    public ConsumerCreator<T> subscriptionInitialPosition(SubscriptionInitialPosition subscriptionInitialPosition) {
        super.subscriptionInitialPosition(subscriptionInitialPosition);
        return this;
    }

    @Override
    public ConsumerCreator<T> subscriptionTopicsMode(RegexSubscriptionMode mode) {
        super.subscriptionTopicsMode(mode);
        return this;
    }

    @Override
    public ConsumerCreator<T> intercept(ConsumerInterceptor<T>... interceptors) {
        super.intercept(interceptors);
        return this;

    }

    @Override
    public ConsumerCreator<T> deadLetterPolicy(DeadLetterPolicy deadLetterPolicy) {
        super.deadLetterPolicy(deadLetterPolicy);
        return this;
    }


    public ConsumerCreator<T> tracing(boolean enableTracing, String projectName) {
        this.enableTracing = enableTracing;
        if (StringUtils.isNotBlank(projectName)) {
            this.projectName = projectName;
        }
        if (traceInterceptor == null) {
            traceInterceptor = new TracingConsumerInterceptor<>(projectName);
            startTracing();
        }
        return this;
    }

    public ConsumerCreator<T> tracing(boolean enableTracing) {
        return tracing(enableTracing, GetAppNameUtil.getAppName());
    }

    protected void startTracing() {
        if (enableTracing && traceInterceptor != null) {
            this.intercept(traceInterceptor);
        }
    }

    public ConsumerCreator<T> metricsCollect(boolean enableMetricsCollect) {
        this.enableMetricsCollect = enableMetricsCollect;
        return this;
    }

    public void initTraceContext(Message<?> message) {
        long spanId = 0L;
        try {
            spanId = Long.parseLong(message.getProperty("spanId"));
        } catch (Exception ignore) {
        }
        boolean sample = true;
        try {
            sample = Boolean.parseBoolean(message.getProperty("sample"));
        } catch (Exception ignore) {
        }
        try {
            traceInterceptor.getTraceContextContainerForAcknowledge()
                    .initZipkinContext(message.getProperty("requestId"), spanId, sample, true);
        } catch (Exception ignore) {
        }
    }

    public TracingConsumerInterceptor<T> getTraceInterceptor() {
        return traceInterceptor;
    }
}
